package com.gjzhao.fmmall.service.impl;

import cn.hutool.json.JSONUtil;
import com.alibaba.excel.util.StringUtils;
import com.gjzhao.fmmall.service.TestService;
import com.gjzhao.fmmall.vo.RespResultVO;
import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import org.elasticsearch.action.DocWriteRequest;
import org.elasticsearch.action.DocWriteResponse;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

/**
 * @Description: 测试 ServiceImpl
 * @Author: gjzhao
 * @Date Created in 2022-01-30 12:43 下午
 */
@Service
public class TestServiceImpl implements TestService {

    private Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired
    private RedissonClient redissonClient; //使用主从模式客户端(不能同时注入)
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    @Autowired
    private RestHighLevelClient restHighLevelClient;

    /**
     * 测试redisson在主从模式下使用分布式锁
     *
     * @param key
     * @return
     */
    @Override
    @Transactional //使用事物保证在本台服务器操作的原子性
    public RespResultVO testRedissonMasterAndSlave(String key) {

        logger.info("---开始测试redisson主从模式下分布式锁的使用-----");

        RespResultVO result = null;
        Map<String, String> map = null;

        RLock fairLock = null;
        try {
            //1.构建非公平锁
            fairLock = redissonClient.getFairLock(key);

            //2.加锁([阻塞锁：如果获取不到锁会尝试再次获取锁]，设置加锁成功后锁的超时时间为20s，防止发生故障时一直不释放锁导致死锁)
            fairLock.lock(20, TimeUnit.SECONDS);

            //3.开始业务操作
            logger.info("---业务执行正常----");
            //测试redis主从是否生效
            stringRedisTemplate.boundValueOps("fmall").set("success_key");

            //4.封装响应
            map = new HashMap();
            map.put(key, "-----业务流程完成-----");

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            logger.info("---释放分布式锁----");
            //释放锁
            fairLock.unlock();
        }
        result = RespResultVO.success(JSONUtil.toJsonStr(map));
        return result;
    }

    /**
     * 测试elasticsearch的基本使用
     *
     * @return
     */
    @Override
    public RespResultVO testElasticSearch() {

        return null;
    }


    /**
     * es批量插入文档
     *
     * @param indexName 索引
     * @param type      类型
     * @param idName    id名称
     * @param list      数据集合
     */
    public void bulkData(String indexName, String type, String idName, List<Map<String, Object>> list) {
        try {

            if (null == list || list.size() <= 0) {
                return;
            }
            if (StringUtils.isBlank(indexName) || StringUtils.isBlank(idName) || StringUtils.isBlank(type)) {
                return;
            }

            BulkRequest request = new BulkRequest();
            for (Map<String, Object> map : list) {
                if (map.get(idName) != null) {
                    request.add(new IndexRequest(indexName, type, String.valueOf(map.get(idName)))
                                    .source(map, XContentType.JSON));
                }
            }

            //可选设置
            request.timeout("2m");
            request.setRefreshPolicy("wait_for");
            request.waitForActiveShards(2);

            //发送同步请求
            BulkResponse bulkResponse = restHighLevelClient.bulk(request, RequestOptions.DEFAULT);

            //处理响应
            if (bulkResponse != null) {
                for (BulkItemResponse bulkItemResponse : bulkResponse) {
                    DocWriteResponse itemResponse = bulkItemResponse.getResponse();
                    if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.INDEX
                        || bulkItemResponse.getOpType() == DocWriteRequest.OpType.CREATE) {
                        IndexResponse indexResponse = (IndexResponse) itemResponse;
                        //todo 新增成功的处理
                        System.out.println("新增成功,{}" + indexResponse.toString());
                    } else if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.UPDATE) {
                        UpdateResponse updateResponse = (UpdateResponse) itemResponse;
                        //todo 修改成功的处理
                        System.out.println("修改成功,{}" + updateResponse.toString());
                    } else if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.DELETE) {
                        DeleteResponse deleteResponse = (DeleteResponse) itemResponse;
                        //todo 删除成功的处理
                        System.out.println("删除成功,{}" + deleteResponse.toString());
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
